package nsqUtils

import (
	"fmt"
	"github.com/nsqio/go-nsq"
	"time"
)

type nsqHandleMessage = func(msg *nsq.Message, err error)

type NsqHandle[T any] struct {
	info    string
	data    T
	handles []nsqHandleMessage
}

func (m *NsqHandle[T]) HandleMessage(msg *nsq.Message) error {
	var err error
	for _, item := range m.handles {
		item(msg, err)
	}
	return err
}
func (m *NsqHandle[T]) SetData(data T) *NsqHandle[T] {
	m.data = data
	return m
}
func (m *NsqHandle[T]) SetMsg(info string) *NsqHandle[T] {
	m.info = info
	return m
}
func NewNsqHandle[T any](data T, handles ...nsqHandleMessage) *NsqHandle[T] {
	return &NsqHandle[T]{data: data, handles: handles}
}

type NsqConsumer struct {
	addr     string
	consumer *nsq.Consumer
}

func NewConsumer(address string) (c *NsqConsumer) {
	return &NsqConsumer{addr: address}
}
func (n *NsqConsumer) InitConsumer(topic string, channel string) *NsqConsumer {
	config := nsq.NewConfig()
	config.LookupdPollInterval = 10 * time.Second
	var err error
	n.consumer, err = nsq.NewConsumer(topic, channel, config)
	if err != nil {
		fmt.Printf("create consumer failed, err:%v\n", err)
		panic(err)
	}
	return n
}

func (n *NsqConsumer) AddHandler(handler ...nsq.Handler) error {
	for _, item := range handler {
		n.consumer.AddHandler(item)
	}
	if err := n.consumer.ConnectToNSQD(n.addr); err != nil { // 直接连NSQD
		//if err := c.ConnectToNSQLookupd(address); err != nil { // 通过lookupd查询
		return err
	}
	return nil
}

func (n *NsqConsumer) AddConcurrentHandlers(handler nsq.Handler, concurrency int) {
	n.consumer.AddConcurrentHandlers(handler, concurrency)
}
